package streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreamingStateSave {
  def main(args: Array[String]): Unit = {
    /**演示无状态数据操作，只对当前周期内的数据进行处理
     * 即像reduceByKey这样的算子
     *采用updateStateByKey()算子可以将当前接受的数据进行状态保留，根据key对数据的状态进行更新，方便下一阶段进行汇总
     *updateStateByKey()：根据key对数据的状态进行更新
     *     第一个参数值：表示相同的key的value数据
     *     第二个参数值：表示缓存区相同key的value数据
     *     使用有状态操作时需要设定检查点路径
     *  */

    val sparkConf=new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    val ssc=new StreamingContext(sparkConf,Seconds(3))
    //流中使用updateStateByKey()时，需要设置point检查点，创建文件夹并保留数据
    ssc.checkpoint("cp")

    val datas: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.80.134", 9999)
    val wordToOne: DStream[(String, Int)] = datas.map((_, 1))
//    val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)
val wordToCount: DStream[(String, Int)] = wordToOne.updateStateByKey(
  (seq: Seq[Int], buffer: Option[Int]) => {
    val newCount = buffer.getOrElse(0) + seq.sum
    Option(newCount)
  }
)
    wordToCount.print()
    ssc.start()
    ssc.awaitTermination()
//    ssc.stop()
    //关闭环境对象
  }

}
